/**
 * File: RuntimeExec.java
 * Created by: mhaimel
 * Created on: 31 Mar 2009
 * CVS:  $Id: RuntimeExec.java,v 1.4 2009/11/24 15:15:40 mhaimel Exp $
 */
package uk.ac.ebi.velvet.exec;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.commons.io.IOUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * Mainly copied from services/services_client/src/main/java/uk/ac/ebi/proteome/util/ProcessExec.java
 * @author mhaimel
 *
 */
public class RuntimeExec {
	/**
	 *
	 */
	private static final String SHELL_FLAGS = "-c";
	/**
	 *
	 */
	private static final String SHELL = "/bin/sh";
	/**
	 * Time to wait in milliseconds for stream gobblers to finish
	 */
	private static final long TIMEOUT = TimeUnit.SECONDS.toMillis(10l);
	
	private Log log = LogFactory.getLog(this.getClass());
	private File baseDir;
	private final ProcessorService pc;
	
	public RuntimeExec(File baseDir) {
		this(baseDir,null);
	}
	
	public RuntimeExec(File baseDir, ProcessorService pc) {
		this.baseDir = baseDir;
		this.pc = pc;
	}
	
	public File getBaseDir() {
		return baseDir;
	}
	
	public int exec(String cmd) throws IOException{
		return execCmd(cmd, null,null);
	}
	
	public int exec(String cmd,StringBuffer out,StringBuffer err) throws IOException{
		ByteArrayOutputStream bO = new ByteArrayOutputStream();
		ByteArrayOutputStream bE = new ByteArrayOutputStream();		
		int ret = execCmd(cmd,bO,bE);
		out.append(bO.toString());
		err.append(bE.toString());
		return ret;
	}
	
	public int exec(String cmd,OutputStream out,OutputStream err) throws IOException{
		return execCmd(cmd,out,err);
	}
	
	public int execArr(String ... cmds) throws IOException{
		return execArrCmd(cmds, null,null);
	}
	
	public int execArr(StringBuffer out,StringBuffer err,String ... cmds) throws IOException{
		ByteArrayOutputStream bO = new ByteArrayOutputStream();
		ByteArrayOutputStream bE = new ByteArrayOutputStream();
		int ret = execArrCmd(cmds,bO,bE);
		out.append(bO.toByteArray());
		err.append(bE.toByteArray());
		return ret;
	}
	public int execArr(OutputStream out,OutputStream err,String ... cmds) throws IOException{
		return execArrCmd( 
				cmds,
				out,err);
	}

	private int execCmd(String cmd, OutputStream out,OutputStream err) throws IOException{
		Runtime rt = Runtime.getRuntime();
		Process proc = rt.exec(new String[] { SHELL, SHELL_FLAGS, cmd}, null,getBaseDir());
		return waitForProcess(out, err, proc);
	}
	
	private int execArrCmd(String[] commandarray, OutputStream out, OutputStream err) throws IOException {
		Runtime rt = Runtime.getRuntime();
		Process proc = rt.exec(commandarray,null,getBaseDir());
		return waitForProcess(out, err, proc);
	}
	
	private int waitForProcess(OutputStream out, OutputStream err, Process proc) {
		if(null != pc){
			pc.add(proc);
		}
		try {
			// kick them off
			submitGobbler(new StreamGobbler(proc.getInputStream(),out, null==out));
			submitGobbler(new StreamGobbler(proc.getErrorStream(),err, null==err));

			// return exit status

			// get the exit status of the command once finished
			int exit = proc.waitFor();
			// now wait for the output and error to be read with a suitable
			// timeout
			join();
			return exit;
		} catch (InterruptedException e) {
			getLog().warn("Process interrupted!");
			Thread.currentThread().interrupt();
			return -1;
		} finally{
			if(null != pc){
				pc.remove(proc);
			}
			// make sure the thread has finished before closing the
			// streams, otherwise the thread tries to read from a closed
			// file and ends up throwing an IOException of "Bad File Descriptor" 
			// and IOException of "Stream closed"
			killQuietlyIfAlive();
			
			// to make sure the all streams are closed to avoid open file handles
			IOUtils.closeQuietly(proc.getErrorStream());
			IOUtils.closeQuietly(proc.getInputStream());
			IOUtils.closeQuietly(proc.getOutputStream());
		}
	}
	
	private Map<StreamGobbler,Future<Object>> gobblerExec = null;
	private List<StreamGobbler> threadExec = null;
	
	private synchronized void submitGobbler(StreamGobbler gobbler) {
		if(null == pc){
			gobbler.start();
			if(null == threadExec){
				threadExec = new ArrayList<StreamGobbler>();
			}
			threadExec.add(gobbler);
		} else {
			Future<Object> future = pc.process((Callable<Object>)gobbler);
			if(null == gobblerExec){
				gobblerExec = new HashMap<StreamGobbler, Future<Object>>();
			}	
			gobblerExec.put(gobbler,future);
		}
	}

	private Log getLog() {
		return log;
	}
	
	private void join(){
		if(null != threadExec){
			for(StreamGobbler sg : threadExec){
				try {
					sg.join(TIMEOUT);
				} catch (InterruptedException e) {
					getLog().error(e);
				}
			}
		} 
		if(null != gobblerExec){
			for(Entry<StreamGobbler, Future<Object>> obj: gobblerExec.entrySet()){
				StreamGobbler gob = obj.getKey();
				Future<Object> future = obj.getValue();
				boolean finished = false;
				int cnt = 0;
				int retry = 10;
				while(!finished && cnt < retry){
					try {
						future.get(TIMEOUT, TimeUnit.MILLISECONDS);
						finished = true;
					} catch (InterruptedException e) {
						getLog().warn(e);
					} catch (ExecutionException e) {
						getLog().warn(e);
					} catch (TimeoutException e) {
						getLog().warn(e);
					} catch (Exception e) {
						getLog().warn(e);
					}
					++ cnt;
				}
				if(!finished){
					getLog().warn("Timeout of "+ (retry*TIMEOUT) + " occured for reading Stream - getting closed ASAP!!!");
					gob.stopASAP();
				}
			}
		}
	}
	
	@SuppressWarnings("deprecation")
	private void killQuietlyIfAlive(){
		if(null != threadExec){
			for(StreamGobbler thread : threadExec){
				if(thread.isAlive()){
					try{
						thread.stopASAP();
						thread.interrupt();
						// last possibility
						thread.join(TIMEOUT);
						if(thread.isAlive()){
							// if still alive -> have to kill it
							thread.stop();
						}
					}catch (Exception e) {
						getLog().error(e);
					}
				}
			}
		} 
		if(null != gobblerExec){
			for(Entry<StreamGobbler, Future<Object>> entry : gobblerExec.entrySet()){
//				StreamGobbler gob = entry.getKey();
				Future<Object> future = entry.getValue();
				try{
					if(future.isDone()){
						// do nothing
					} else if(future.isCancelled()){
						// do nothing
					} else{
						future.cancel(true);
						getLog().warn("Cancel collecting stream after timeout!!!");
//						throw new VelvetUncheckedException(
//								"Gobbler has not finished at this point - need to inforce finishing!");
					}
				}catch (Exception e) {
					getLog().warn(e);
				}
			}
		}
	}
}


/**
 * Class to consume a stream from an executing process to avoid lockups. Code
 * derived but extensively modified from <a
 * href="http://www.javaworld.com/javaworld/jw-12-2000/jw-1229-traps.html">JavaWorld
 * article</a>
 *
 * @author mhaimel
 *
 */
class StreamGobbler extends Thread implements Callable<Object>{

	private static final int DEFAULT_BUFFER_SIZE = 1024;
	private Log log = LogFactory.getLog(this.getClass());
	final AtomicReference<InputStream> is = new AtomicReference<InputStream>();
	final AtomicReference<OutputStream> out = new AtomicReference<OutputStream>();

	final AtomicBoolean discard = new AtomicBoolean(false);
	final AtomicBoolean stop = new AtomicBoolean(false);


	StreamGobbler(InputStream is, OutputStream out, boolean discard) {
		this.is.set(is);
		this.out.set(out);
		this.discard.set(discard);
	}

	public void run() {
		try {
			if (discard.get()) {
				steamToDevNull();
			} else {
				streamToBuffer();
			}
		} catch (Exception ioe) {
			log.warn("Problems reading process output!",ioe);
		} finally {
			IOUtils.closeQuietly(is.getAndSet(null));
		}
	}

	private void steamToDevNull() throws IOException {
		final byte[] buf = new byte[getBufferSize()];
        while (!gotInterrupted() && !stop.get() && (is.get().read(buf) > 0)) {
        }
	}

	private int getBufferSize() {
		return DEFAULT_BUFFER_SIZE;
	}

	private void streamToBuffer()throws IOException {
		final byte[] buf = new byte[getBufferSize()];
        int length = 0;
        while (!gotInterrupted() && !stop.get() && ((length = is.get().read(buf)) > 0) ) {
            out.get().write(buf, 0, length);
        }
	}
	
	public boolean gotInterrupted(){
		if(isInterrupted()){
			return true;
		}
		if(Thread.currentThread().isInterrupted()){
			return true;
		}
		return false;
	}
	
	public void stopASAP(){
		this.stop.set(true);
	}

	@Override
	public Object call() throws Exception {
		run();
		return "Finished";
	}
}
